package com.yixing.tech.common.tcp.pc;

import com.yixing.tech.common.tcp.channel.ChannelContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;

import java.util.UUID;
import java.util.concurrent.BlockingDeque;
import java.util.function.Function;

@Slf4j
public class TcpConsumer implements Runnable {

    private boolean runnable = true;

    private boolean flag = true;

    private ChannelContext channelContext;

    private final BlockingDeque<TcpMessage> deque;

    private final Function<TcpMessage, TcpReply> func;

    public TcpConsumer(BlockingDeque<TcpMessage> deque, Function<TcpMessage, TcpReply> func) {
        this.deque = deque;
        this.func = func;
    }

    public TcpConsumer(BlockingDeque<TcpMessage> deque, Function<TcpMessage, TcpReply> func, ChannelContext channelContext) {
        this.deque = deque;
        this.func = func;
        this.channelContext = channelContext;
    }

    public void setChannelContext(ChannelContext channelContext) {
        this.channelContext = channelContext;
    }

    @Override
    public void run() {
        while(runnable) {
            try {
                if(flag) {
                    for(int i=0; i<deque.size(); ++i) {
                        String uid = UUID.randomUUID().toString().replaceAll("-", "");
                        TcpMessage message = deque.takeFirst();
//                        log.debug("consume msg start:message[{}],uid[{}]", message, uid);

                        TcpReply reply = this.func.apply(message);
                        if(ObjectUtils.isNotEmpty(reply) && ObjectUtils.isNotEmpty(channelContext)) {
                            channelContext.sendDataByChannel(reply.getIp(), reply.getMessage());
                        }

//                        log.debug("consume msg end:message[{}],uid[{}]", message, uid);
                    }
                }
                Thread.sleep(10);
            } catch (InterruptedException e) {
                log.error("tcp消费者消费数据发生异常:{}", e.getMessage());
            }
        }
    }

    public void stop() {
        this.runnable = false;
    }

    public void resume() {
        this.flag = true;
    }

    public void suspend() {
        this.flag = false;
    }
}
